package cn.waqwb;

import org.apache.hadoop.conf.Configuration;

import java.net.URI;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.io.IOUtils;

/**
 * @Classname HDFSOperateDemo
 * @Description TODO
 * @Date 2022/9/20 19:14
 * @Created by Administrator
 */
public class HDFSOperateDemo {
    public static void main(String[] args) throws Exception {
//        getFileSystem();
//		uploadFile();
        //createFile();
//        createDir();
        deleteFile();
        //fileRename();
    }

    /**
     * 构建获取FileSystem接口API的方法供以后的文件操作使用
     * @return
     * @throws Exception
     */
    static FileSystem getFileSystem() throws Exception{
        URI uri = new URI("hdfs://192.168.137.131:9000/");
        FileSystem fileSystem = FileSystem.get(uri, new Configuration(),"root");
        return fileSystem;
    }

    /**
     * 使用FileSystem的
     * copyFromLocalFile（本地文件，HDFS路径）方法将本地文件传到HDFS中
     * @throws Exception
     */
    public static void uploadFile() throws Exception {
        //通过FileSystem类和HDFS建立连接
        FileSystem hdfs = getFileSystem();
        //文件源路径
        Path src = new Path("./data/file2.txt");
        //目标路径
        Path dst = new Path("\\test1");
        //通过调用listStatus方法，把路径下所有文件存到数组中（上传前）
        FileStatus files[] = hdfs.listStatus(dst);
        //遍历数组
        for (FileStatus file : files) {
            System.out.println(file.getPath());
        }
        System.out.println("------------上传之后--------------------");
        //调用copyFromLocalFile把本地文件上传到HDFS中
        hdfs.copyFromLocalFile(src, dst);
        //通过调用listStatus方法，把路径下所有文件存到数组中（上传后）
        files = hdfs.listStatus(dst);
        for (FileStatus file : files) {
            System.out.println(file.getPath());
        }
    }

    /**
     * 在HDFS上新建文件，并写入相应文字
     * @throws Exception
     */
    public static void createFile() throws Exception{
        //通过FileSystem类和HDFS建立连接
        FileSystem hdfs = getFileSystem();
        //准备好要写入的文字
        byte[] buff = "HelloHadoop".getBytes();
        //在HDFS上创建文件，路径需要写完整路径
        Path path = new Path("/test2/Hello.txt");
        //FSDataOutputStream重载了很多write方法，用于写入很多类型的数据
        FSDataOutputStream output = hdfs.create(path);
        //写入数据
        output.write(buff,0,buff.length);
        System.out.println("Finish....");
        output.close();
    }

    /**
     * 为指定的HDFS文件重命名
     * 其中src和dst均为文件的完整路径
     * @throws Exception
     */
    public static void fileRename() throws Exception {
        //通过FileSystem类和HDFS建立连接
        FileSystem hdfs = getFileSystem();
        //原文件名，                       src
        Path frpath = new Path("/test1/1.txt");
        //新文件名，                       dst
        Path topath = new Path("/test1/2.txt");
        //重命名
        boolean isRename = hdfs.rename(frpath, topath);
        String result = isRename ? "成功" : "失败";
        System.out.println("文件重命名结果为：" + result);
    }

    /**
     * 使用mkdirs创建目录
     * @throws Exception
     */
    public static void createDir() throws Exception {
        //通过FileSystem类和HDFS建立连接
        FileSystem hdfs = getFileSystem();
        //切换到目录test4下
        Path path = new Path("/test");
        //mkdirs创建目录
        boolean flag = hdfs.mkdirs(path);
        //判断是否成功
        String str = flag?"成功":"失败";
        System.out.println("创建文件夹"+str);
    }

    /**
     * 通过FileSystem.delete(Path f,Boolean recursive)可删除指定的HDFS文件，
     * 其中f为需要删除文件的完整路径，recursive用来确定是否进行递归删除
     * @throws Exception
     */
    public static void deleteFile() throws Exception{
        FileSystem hdfs = getFileSystem();
        Path path = new Path("/test");
        String str = hdfs.delete(path,true)?"成功":"失败";
        System.out.println("删除文件"+str);
    }

    /**
     *  打开指定的HDFS文件进行读取
     * @throws Exception
     */
    public static void readFile() throws Exception {
        FileSystem fileSystem = getFileSystem();
        FSDataInputStream openStream = fileSystem.open(new Path("/testcreate"));
        //import org.apache.hadoop.io.IOUtils;
        IOUtils.copyBytes(openStream, System.out, 1024, false);
        IOUtils.closeStream(openStream);
    }

    /**
     *  查看指定的HDFS文件是否存在
     * @throws Exception
     */
    public static void isFileExists() throws Exception {
        FileSystem hdfs = getFileSystem();
        Path findf = new Path("/test1");
        boolean isExists = hdfs.exists(findf);
        if (isExists == false){
            System.out.print("不");
        }
        System.out.println("存在");
    }

    /**
     *  查看指定的HDFS文件的修改时间
     * @throws Exception
     */
    public static void fileLastModify() throws Exception {
        FileSystem hdfs = getFileSystem();
        Path fpath = new Path("/testcreate");
        FileStatus fileStatus = hdfs.getFileStatus(fpath);
        long modiTime = fileStatus.getModificationTime();
        System.out.println("testcreate的修改时间是" + modiTime);
    }
    /**
     *  查看指定的HDFS中某个目录下的所有文件
     * @throws Exception
     */
    public static void fileLocation() throws Exception {
        FileSystem hdfs = getFileSystem();
        Path fpath = new Path("/testcreate");
        FileStatus filestatus = hdfs.getFileStatus(fpath);
        BlockLocation[] blkLocations = hdfs.getFileBlockLocations(filestatus, 0, filestatus.getLen());
        int blockLen = blkLocations.length;
        for (int i = 0; i < blockLen; i++) {
            String[] hosts = blkLocations[i].getHosts();
            System.out.println("block_" + i + "_location:" + hosts[0]);
        }
    }

    /**
     *  获取HDFS集群上的所有节点名称
     * @throws Exception
     */
    public static void nodeList() throws Exception {
        FileSystem fs = getFileSystem();
        DistributedFileSystem hdfs = (DistributedFileSystem) fs;

        DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
        for (int i = 0; i < dataNodeStats.length; i++) {
            System.out.println("DataNode_" + i + "_Name:"
                    + dataNodeStats[i].getHostName());
        }
    }
}
